from pyspark.sql import SparkSession
import os

"""
-------------------------------------------------
   Description :   TODO：测试Spark连接MySQL数据库
   SourceFile  :   Demo01_TestMySQL
   Author      :   itcast team
-------------------------------------------------
"""

# 0.设置系统环境变量 - 配置Spark运行所需的环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'  # 设置JDK安装路径
os.environ['SPARK_HOME'] = '/export/server/spark'  # 设置Spark安装路径
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'  # 设置Python执行环境
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'  # 设置Driver端Python环境

# 1.构建SparkSession对象
# 使用建造者模式创建SparkSession，配置Spark应用的基本参数
# spark = SparkSession \
#     .builder \
#     .master("local[2]") \  # 本地模式运行，使用2个线程
#     .appName("SparkSQLAppName") \  # 设置应用名称
#     .config("spark.sql.shuffle.partitions", 4) \  # 设置shuffle分区数为4
#     .getOrCreate()  # 获取或创建SparkSession
spark = SparkSession \
    .builder \
    .master("local[2]") \
    .appName("SparkSQLAppName") \
    .config("spark.sql.shuffle.partitions", 4) \
    .getOrCreate()

# 2.从MySQL数据库读取数据
# 使用jdbc方式连接MySQL，读取edu数据库中的base_region表
input_df = spark.read.jdbc(
    url='jdbc:mysql://up01:3306/edu',  # MySQL连接URL
    table='base_region',  # 要读取的表名
    properties={'user':'root','password':'123456'}  # 数据库用户名和密码
)

# 3.数据处理和展示
input_df.printSchema()  # 打印DataFrame的schema结构
input_df.show()  # 展示DataFrame中的数据内容

# 4.将处理后的数据保存回MySQL数据库
# 将DataFrame写入到edu数据库的base_region42表中
input_df.write.jdbc(
    url='jdbc:mysql://up01:3306/edu',  # MySQL连接URL
    table='base_region42',  # 目标表名
    mode='overwrite',  # 写入模式：覆盖已有数据
    properties={'user':'root','password':'123456'}  # 数据库用户名和密码
)

# 5.关闭SparkSession，释放资源
spark.stop()
